package com.fullertontech.framework.rabbitmq;

import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;

import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import com.rabbitmq.client.Channel;

@Component
public final class MessageListener implements ChannelAwareMessageListener,org.springframework.amqp.core.MessageListener,InitializingBean,ApplicationContextAware,BeanPostProcessor{
	@Autowired
	private CachingConnectionFactory cachingConnectionFactory;

	private RabbitClient rabbitClient;
	
	private AcknowledgeMode acknowledgeMode; 
	
	private ApplicationContext applicationContext;
	
	private volatile int messageListenerContainerIndex=0;
	
	public void setAcknowledgeMode(AcknowledgeMode acknowledgeMode) {
		this.acknowledgeMode = acknowledgeMode;
	}

	@Override
	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
		this.applicationContext=applicationContext;
	}

	public void setRabbitClient(RabbitClient rabbitClient) {
		this.rabbitClient = rabbitClient;
	}

	@Override
	public synchronized void afterPropertiesSet() throws Exception {
		acknowledgeMode=AcknowledgeMode.AUTO;
	}
	/**
	 * 必须实现org.springframework.amqp.core.MessageListener，否则会报错
	 */
	@Override
	public final void onMessage(Message message) {
		
	}
	@Override
	public final void onMessage(Message message,Channel channel) {
		try {
			Class<?> listenerClass=rabbitListenerClass.get(message.getMessageProperties().getConsumerQueue());
			Object listener=applicationContext.getBean(listenerClass);
			Method method=listenerClass.getMethod("onMessage", new Class[] {String.class});
			method.invoke(listener,new String(message.getBody()));
			if(acknowledgeMode!=null&&acknowledgeMode.equals(AcknowledgeMode.MANUAL)) {//正常消费掉，队列被确认和丢弃
				channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
			}
		} catch (Exception e) {
			e.printStackTrace();
			try {
				if(acknowledgeMode!=null&&acknowledgeMode.equals(AcknowledgeMode.MANUAL)) {//业务出现异常消息被重新扔在队列中
					channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
					channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
				}
			} catch (Exception e1) {
				e1.printStackTrace();
			}
		}
	}
	
	private static Map<String,Class<?>> rabbitListenerClass=new HashMap<>();
	
	public static Map<String,Class<?>> getRabbitListenerClass() {
		return rabbitListenerClass;
	}
	
	@Override
	public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
		RabbitListener rabbitListener=bean.getClass().getAnnotation(RabbitListener.class);
		if(rabbitListener!=null) {
			String queueName=rabbitListener.queue();
			String exchange=rabbitListener.exchange();
			if(StringUtils.isEmpty(exchange)) {
				exchange=rabbitListener.queue();
			}
			try {
				rabbitClient.createBindQueue(queueName, exchange, rabbitListener.exchangeTypes());
			} catch (Exception e) {
				e.printStackTrace();
			}
			rabbitListenerClass.put(rabbitListener.queue(),bean.getClass());
			DefaultListableBeanFactory factory=(DefaultListableBeanFactory)applicationContext.getAutowireCapableBeanFactory();
	        BeanDefinitionBuilder builder=BeanDefinitionBuilder.genericBeanDefinition(SimpleMessageListenerContainer.class);
	        builder.addPropertyValue("connectionFactory", cachingConnectionFactory);
	        builder.addPropertyValue("maxConcurrentConsumers", Integer.parseInt(rabbitListener.maxConcurrentConsumers()));
	        builder.addPropertyValue("concurrentConsumers", Integer.parseInt(rabbitListener.maxConcurrentConsumers()));
	        builder.addPropertyValue("acknowledgeMode", AcknowledgeMode.AUTO);
	        builder.addPropertyValue("messageListener", this);
	        builder.addPropertyValue("prefetchCount", 1);
	        builder.addPropertyValue("queueNames", queueName);
	        factory.registerBeanDefinition(SimpleMessageListenerContainer.class.getName()+(messageListenerContainerIndex++), builder.getBeanDefinition());
	        return bean;  
		}
		return bean;
	}

	@Override
	public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
		return bean;
	}

}
